"""Data Quality Service - 数据质量监控与自动修复(简化版).

根据审计报告,该模块原完成度仅28%,现实现核心功能。
包含数据标准化工具函数用于导入审核流程。
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Optional
import logging
import re

from sqlalchemy import func
from sqlalchemy.orm import Session
from pypinyin import lazy_pinyin, Style
from rapidfuzz.distance import Levenshtein

from ..models.lead import Lead
from ..models.property import Property
from ..models.user import User

logger = logging.getLogger(__name__)


def normalize_phone(phone: str | None) -> str:
    """
    标准化手机号码.

    处理规则:
    - 去除空格、横线、括号等分隔符
    - 去除国家代码前缀 (+86, 0086, 86-)
    - 保留最后11位本地号

    示例:
        "138-0013-8000" -> "13800138000"
        "+86 138 0013 8000" -> "13800138000"
        "0086-13800138000" -> "13800138000"

    Args:
        phone: 原始电话号码

    Returns:
        str: 标准化后的11位手机号 (若无效则返回原字符串)
    """
    if not phone:
        return ""

    # 去除所有非数字字符
    digits_only = re.sub(r"\D", "", phone)

    # 去除国家代码前缀
    if digits_only.startswith("86") and len(digits_only) > 11:
        digits_only = digits_only[2:]
    elif digits_only.startswith("0086"):
        digits_only = digits_only[4:]

    # 保留最后11位 (中国手机号标准长度)
    if len(digits_only) > 11:
        digits_only = digits_only[-11:]

    return digits_only


def to_pinyin(chinese_text: str) -> str:
    """
    将中文转换为拼音 (无音调).

    示例:
        "张三" -> "zhangsan"
        "王伟" -> "wangwei"

    Args:
        chinese_text: 中文文本

    Returns:
        str: 拼音 (小写,无分隔符)
    """
    if not chinese_text:
        return ""

    # 使用pypinyin库转换,Style.NORMAL表示无音调
    pinyin_list = lazy_pinyin(chinese_text, style=Style.NORMAL)
    return "".join(pinyin_list).lower()


def normalize_name(name: str | None) -> dict[str, str]:
    """
    标准化姓名.

    处理规则:
    - 去除前后空白
    - 统一英文大小写 (首字母大写)
    - 中文名转拼音 (用于模糊匹配)

    Args:
        name: 原始姓名

    Returns:
        dict: {
            "original": 原始姓名 (清理后),
            "normalized": 标准化姓名,
            "pinyin": 拼音形式 (仅中文名)
        }
    """
    if not name:
        return {"original": "", "normalized": "", "pinyin": ""}

    # 去除前后空白和特殊字符
    cleaned = name.strip()
    cleaned = re.sub(r"[/\\|]", " ", cleaned)
    cleaned = re.sub(r"\s+", " ", cleaned)

    # 判断是否包含中文
    has_chinese = bool(re.search(r"[\u4e00-\u9fff]", cleaned))

    if has_chinese:
        # 中文名: 转换拼音用于匹配
        pinyin_form = to_pinyin(cleaned)
        normalized = cleaned
    else:
        # 英文名: 标准化大小写 (每个单词首字母大写)
        normalized = cleaned.title()
        pinyin_form = normalized.lower().replace(" ", "")

    return {"original": cleaned, "normalized": normalized, "pinyin": pinyin_form}


def normalize_area(
    area: str | None, area_dict: dict[str, str] | None = None
) -> dict[str, str | bool | float | list[str]]:
    """
    标准化区域/片区名称.

    使用内部字典进行映射匹配,并提取行政区信息.

    Args:
        area: 原始区域名称
        area_dict: 区域映射字典 {别名: 标准名}, 默认使用内置字典

    Returns:
        dict: {
            "original": 原始区域名,
            "normalized": 标准化区域名,
            "matched": 是否匹配到字典,
            "confidence": 匹配置信度,
            "districts": 提取的行政区列表
        }
    """
    if not area:
        return {
            "original": "",
            "normalized": "",
            "matched": False,
            "confidence": 0.0,
            "districts": [],
        }

    # 去除空白
    cleaned = area.strip()

    # 默认区域字典 (示例 - 实际应从配置文件或数据库加载)
    if area_dict is None:
        area_dict = {
            # 片区别名 -> 标准名
            "CBD": "中央商务区",
            "cbd": "中央商务区",
            "市中心": "中心城区",
            "高新区": "高新技术开发区",
            "浦东": "浦东新区",
            "浦西": "浦西片区",
            "朝阳": "朝阳区",
            "海淀": "海淀区",
            "三里屯": "朝阳区三里屯",
            # 更多映射...
        }

    # 提取行政区 (简单模式匹配)
    districts = []
    common_districts = [
        "朝阳区",
        "海淀区",
        "东城区",
        "西城区",
        "丰台区",
        "石景山区",
        "浦东新区",
        "黄浦区",
        "静安区",
        "徐汇区",
        "长宁区",
        "普陀区",
    ]

    for district in common_districts:
        if district in cleaned or district.replace("区", "") in cleaned:
            districts.append(district)
            break

    # 精确匹配
    if cleaned in area_dict:
        return {
            "original": cleaned,
            "normalized": area_dict[cleaned],
            "matched": True,
            "confidence": 1.0,
            "districts": districts,
        }

    # 模糊匹配 (简单包含检查)
    for alias, standard in area_dict.items():
        if alias.lower() in cleaned.lower() or cleaned.lower() in alias.lower():
            return {
                "original": cleaned,
                "normalized": standard,
                "matched": True,
                "confidence": 0.8,
                "districts": districts,
            }

    # 无匹配,返回原值
    return {
        "original": cleaned,
        "normalized": cleaned,
        "matched": False,
        "confidence": 0.0,
        "districts": districts,
    }


def calculate_phone_similarity(phone1: str, phone2: str) -> float:
    """
    计算手机号相似度.

    规则: 标准化后完全一致 = 1.0, 否则 = 0.0

    Args:
        phone1: 手机号1
        phone2: 手机号2

    Returns:
        float: 相似度分数 (0.0 或 1.0)
    """
    norm1 = normalize_phone(phone1)
    norm2 = normalize_phone(phone2)

    if not norm1 or not norm2:
        return 0.0

    return 1.0 if norm1 == norm2 else 0.0


def calculate_name_similarity(name1: str, name2: str) -> float:
    """
    计算姓名相似度 (支持中文/英文/拼音匹配).

    使用Levenshtein编辑距离算法:
    - 完全一致: 1.0
    - 拼音一致: 0.8-0.95
    - 部分相似: 0.3-0.8

    Args:
        name1: 姓名1
        name2: 姓名2

    Returns:
        float: 相似度分数 (0.0 - 1.0)
    """
    if not name1 or not name2:
        return 0.0

    # 标准化姓名
    norm1 = normalize_name(name1)
    norm2 = normalize_name(name2)

    n1_orig = norm1["normalized"]
    n2_orig = norm2["normalized"]
    n1_pinyin = norm1["pinyin"]
    n2_pinyin = norm2["pinyin"]

    # 1. 精确匹配 (标准化后)
    if n1_orig == n2_orig:
        return 1.0

    # 2. 拼音完全匹配
    if n1_pinyin and n2_pinyin and n1_pinyin == n2_pinyin:
        return 0.95

    # 3. 计算Levenshtein相似度
    max_len = max(len(n1_orig), len(n2_orig))
    if max_len == 0:
        return 0.0

    # 原始姓名编辑距离
    dist_orig = Levenshtein.distance(n1_orig, n2_orig)
    similarity_orig = 1.0 - (dist_orig / max_len)

    # 拼音编辑距离 (如果有)
    if n1_pinyin and n2_pinyin:
        max_len_pinyin = max(len(n1_pinyin), len(n2_pinyin))
        dist_pinyin = Levenshtein.distance(n1_pinyin, n2_pinyin)
        similarity_pinyin = 1.0 - (dist_pinyin / max_len_pinyin)

        # 取两者最高值 (拼音匹配可能更准确)
        return max(similarity_orig, similarity_pinyin * 0.9)

    return similarity_orig


def calculate_area_similarity(area1: str, area2: str) -> float:
    """
    计算区域相似度.

    简单字符串匹配:
    - 完全一致: 1.0
    - 部分包含: 0.5
    - 不相关: 0.0

    Args:
        area1: 区域1
        area2: 区域2

    Returns:
        float: 相似度分数 (0.0 - 1.0)
    """
    if not area1 or not area2:
        return 0.0

    a1 = area1.strip().lower()
    a2 = area2.strip().lower()

    if a1 == a2:
        return 1.0

    # 检查包含关系
    if a1 in a2 or a2 in a1:
        return 0.5

    return 0.0


def calculate_duplicate_score(
    phone1: str,
    name1: str,
    area1: str,
    phone2: str,
    name2: str,
    area2: str,
    weights: tuple[float, float, float] = (0.6, 0.3, 0.1),
) -> dict[str, Any]:
    """
    计算记录重复得分 (综合phone/name/area).

    公式: score = w1*phone_score + w2*name_score + w3*area_score

    阈值分类:
    - >= 0.90: 强建议合并
    - 0.75-0.90: 推荐合并 (需人工确认)
    - < 0.75: 不建议合并

    Args:
        phone1, name1, area1: 记录1的字段
        phone2, name2, area2: 记录2的字段
        weights: 权重 (phone, name, area), 默认 (0.6, 0.3, 0.1)

    Returns:
        dict: {
            "score": 综合得分,
            "phone_score": 手机号得分,
            "name_score": 姓名得分,
            "area_score": 区域得分,
            "category": 分类 ("high"/"medium"/"low"),
            "recommendation": 建议
        }
    """
    w_phone, w_name, w_area = weights

    phone_score = calculate_phone_similarity(phone1, phone2)
    name_score = calculate_name_similarity(name1, name2)
    area_score = calculate_area_similarity(area1, area2)

    # 计算综合得分
    combined_score = w_phone * phone_score + w_name * name_score + w_area * area_score

    # 分类
    if combined_score >= 0.90:
        category = "high"
        recommendation = "auto_merge"
    elif combined_score >= 0.75:
        category = "medium"
        recommendation = "review_merge"
    else:
        category = "low"
        recommendation = "ignore"

    return {
        "score": combined_score,
        "phone_score": phone_score,
        "name_score": name_score,
        "area_score": area_score,
        "category": category,
        "recommendation": recommendation,
    }


@dataclass
class QualityIssue:
    """数据质量问题."""

    severity: str  # "critical", "high", "medium", "low"
    category: str  # "duplicate", "missing", "invalid", "stale"
    entity_type: str  # "lead", "property", "user"
    entity_id: int
    field_name: str
    issue_description: str
    suggested_fix: Optional[str] = None
    auto_fixable: bool = False


@dataclass
class QualityReport:
    """数据质量报告."""

    scan_time: datetime
    total_records: int
    issues_count: int
    critical_count: int
    high_count: int
    medium_count: int
    low_count: int
    issues: list[QualityIssue]
    auto_fixed_count: int = 0


class DataQualityService:
    """数据质量监控服务 (Synchronous)."""

    def __init__(self, db: Session):
        self.db = db

    def scan_all(self, auto_fix: bool = False) -> QualityReport:
        """
        全面扫描数据质量问题.

        Args:
            auto_fix: 是否自动修复可修复的问题

        Returns:
            QualityReport: 质量报告
        """
        issues: list[QualityIssue] = []

        # 扫描各个实体
        issues.extend(self.scan_leads())
        issues.extend(self.scan_properties())
        issues.extend(self.scan_users())

        # 统计问题
        severity_counts = {
            "critical": sum(1 for i in issues if i.severity == "critical"),
            "high": sum(1 for i in issues if i.severity == "high"),
            "medium": sum(1 for i in issues if i.severity == "medium"),
            "low": sum(1 for i in issues if i.severity == "low"),
        }

        # 自动修复
        auto_fixed_count = 0
        if auto_fix:
            auto_fixed_count = self.auto_fix_issues(issues)

        # 计算总记录数
        total_records = (
            self.db.query(func.count(Lead.id)).scalar()
            + self.db.query(func.count(Property.id)).scalar()
            + self.db.query(func.count(User.id)).scalar()
        )

        return QualityReport(
            scan_time=datetime.now(),
            total_records=total_records or 0,
            issues_count=len(issues),
            critical_count=severity_counts["critical"],
            high_count=severity_counts["high"],
            medium_count=severity_counts["medium"],
            low_count=severity_counts["low"],
            issues=issues,
            auto_fixed_count=auto_fixed_count,
        )

    def scan_leads(self) -> list[QualityIssue]:
        """扫描客户数据质量."""
        issues: list[QualityIssue] = []

        # 1. 检查重复手机号
        duplicates = (
            self.db.query(Lead.phone, func.count(Lead.id).label("count"))
            .group_by(Lead.phone)
            .having(func.count(Lead.id) > 1)
            .all()
        )

        for phone, count in duplicates:
            duplicate_leads = self.db.query(Lead).filter(Lead.phone == phone).all()
            for lead in duplicate_leads[1:]:  # 保留第一个,标记其他为重复
                issues.append(
                    QualityIssue(
                        severity="high",
                        category="duplicate",
                        entity_type="lead",
                        entity_id=lead.id,
                        field_name="phone",
                        issue_description=f"重复手机号: {phone} (共{count}条记录)",
                        suggested_fix=f"合并或删除重复记录 (ID: {lead.id})",
                        auto_fixable=False,  # 需要人工确认
                    )
                )

        # 2. 检查无效手机号格式
        leads = self.db.query(Lead).all()
        for lead in leads:
            if not lead.phone or len(lead.phone) != 11 or not lead.phone.isdigit():
                issues.append(
                    QualityIssue(
                        severity="medium",
                        category="invalid",
                        entity_type="lead",
                        entity_id=lead.id,
                        field_name="phone",
                        issue_description=f"无效手机号格式: {lead.phone}",
                        suggested_fix="修正为11位数字手机号",
                        auto_fixable=False,
                    )
                )

        # 3. 检查空白姓名
        empty_name_leads = (
            self.db.query(Lead).filter((Lead.name.is_(None)) | (Lead.name == "")).all()
        )

        for lead in empty_name_leads:
            issues.append(
                QualityIssue(
                    severity="medium",
                    category="missing",
                    entity_type="lead",
                    entity_id=lead.id,
                    field_name="name",
                    issue_description="客户姓名为空",
                    suggested_fix=f"自动填充为 '客户{lead.phone[-4:]}'",
                    auto_fixable=True,
                )
            )

        # 4. 检查长时间未跟进的客户 (30天以上)
        stale_threshold = datetime.now() - timedelta(days=30)
        stale_leads = (
            self.db.query(Lead)
            .filter(
                Lead.updated_at < stale_threshold,
                Lead.stage != "closed_won",
                Lead.stage != "closed_lost",
            )
            .all()
        )

        for lead in stale_leads:
            days_stale = (datetime.now() - lead.updated_at).days
            issues.append(
                QualityIssue(
                    severity="low",
                    category="stale",
                    entity_type="lead",
                    entity_id=lead.id,
                    field_name="updated_at",
                    issue_description=f"客户{days_stale}天未更新",
                    suggested_fix="提醒经纪人跟进或标记为流失",
                    auto_fixable=False,
                )
            )

        return issues

    def scan_properties(self) -> list[QualityIssue]:
        """扫描房源数据质量."""
        issues: list[QualityIssue] = []

        # 1. 检查重复房源 (同小区+同房号)
        duplicates = (
            self.db.query(
                Property.community_name,
                Property.room_number,
                func.count(Property.id).label("count"),
            )
            .group_by(Property.community_name, Property.room_number)
            .having(func.count(Property.id) > 1)
            .all()
        )

        for community, room, count in duplicates:
            duplicate_props = (
                self.db.query(Property)
                .filter(Property.community_name == community, Property.room_number == room)
                .all()
            )

            for prop in duplicate_props[1:]:
                issues.append(
                    QualityIssue(
                        severity="high",
                        category="duplicate",
                        entity_type="property",
                        entity_id=prop.id,
                        field_name="community_name,room_number",
                        issue_description=f"重复房源: {community} {room} (共{count}条)",
                        suggested_fix=f"合并或删除重复房源 (ID: {prop.id})",
                        auto_fixable=False,
                    )
                )

        # 2. 检查异常租金 (过高或过低)
        MIN_RENT = 500
        MAX_RENT = 50000

        abnormal_price_props = (
            self.db.query(Property)
            .filter((Property.rent_price < MIN_RENT) | (Property.rent_price > MAX_RENT))
            .all()
        )

        for prop in abnormal_price_props:
            issues.append(
                QualityIssue(
                    severity="medium",
                    category="invalid",
                    entity_type="property",
                    entity_id=prop.id,
                    field_name="rent_price",
                    issue_description=f"异常租金: {prop.rent_price}元/月 (正常范围: {MIN_RENT}-{MAX_RENT})",
                    suggested_fix="检查是否录入错误",
                    auto_fixable=False,
                )
            )

        # 3. 检查空白房东信息
        empty_landlord_props = (
            self.db.query(Property)
            .filter((Property.landlord_phone.is_(None)) | (Property.landlord_phone == ""))
            .all()
        )

        for prop in empty_landlord_props:
            issues.append(
                QualityIssue(
                    severity="medium",
                    category="missing",
                    entity_type="property",
                    entity_id=prop.id,
                    field_name="landlord_phone",
                    issue_description="房东电话为空",
                    suggested_fix="补充房东联系方式",
                    auto_fixable=False,
                )
            )

        return issues

    def scan_users(self) -> list[QualityIssue]:
        """扫描用户数据质量."""
        issues: list[QualityIssue] = []

        # 1. 检查未激活用户
        inactive_users = self.db.query(User).filter(User.is_active.is_(False)).all()

        for user in inactive_users:
            issues.append(
                QualityIssue(
                    severity="low",
                    category="stale",
                    entity_type="user",
                    entity_id=user.id,
                    field_name="is_active",
                    issue_description=f"用户未激活: {user.email}",
                    suggested_fix="发送激活邮件或删除账户",
                    auto_fixable=False,
                )
            )

        # 2. 检查缺少姓名的用户
        empty_name_users = (
            self.db.query(User).filter((User.full_name.is_(None)) | (User.full_name == "")).all()
        )

        for user in empty_name_users:
            issues.append(
                QualityIssue(
                    severity="low",
                    category="missing",
                    entity_type="user",
                    entity_id=user.id,
                    field_name="full_name",
                    issue_description="用户姓名为空",
                    suggested_fix=f"自动填充为 '{user.email.split('@')[0]}'",
                    auto_fixable=True,
                )
            )

        return issues

    def auto_fix_issues(self, issues: list[QualityIssue]) -> int:
        """
        自动修复可修复的问题.

        Args:
            issues: 质量问题列表

        Returns:
            int: 修复的问题数量
        """
        fixed_count = 0

        for issue in issues:
            if not issue.auto_fixable:
                continue

            try:
                if issue.entity_type == "lead" and issue.field_name == "name":
                    # 修复空白客户姓名
                    lead = self.db.query(Lead).filter(Lead.id == issue.entity_id).first()
                    if lead and (not lead.name or lead.name == ""):
                        lead.name = f"客户{lead.phone[-4:]}"
                        self.db.flush()
                        fixed_count += 1
                        logger.info(f"自动修复: Lead {lead.id} 姓名填充为 {lead.name}")

                elif issue.entity_type == "user" and issue.field_name == "full_name":
                    # 修复空白用户姓名
                    user = self.db.query(User).filter(User.id == issue.entity_id).first()
                    if user and (not user.full_name or user.full_name == ""):
                        user.full_name = user.email.split("@")[0]
                        self.db.flush()
                        fixed_count += 1
                        logger.info(f"自动修复: User {user.id} 姓名填充为 {user.full_name}")

            except Exception as e:
                logger.error(f"自动修复失败: {issue.entity_type} {issue.entity_id}, 错误: {str(e)}")

        if fixed_count > 0:
            self.db.commit()

        return fixed_count

    def get_quality_summary(self) -> dict[str, Any]:
        """
        获取数据质量概览 (快速统计).

        Returns:
            dict: 质量统计摘要
        """
        total_leads = self.db.query(func.count(Lead.id)).scalar() or 0
        total_properties = self.db.query(func.count(Property.id)).scalar() or 0
        total_users = self.db.query(func.count(User.id)).scalar() or 0

        # 快速检查关键问题
        duplicate_phones = (
            self.db.query(func.count())
            .select_from(
                self.db.query(Lead.phone)
                .group_by(Lead.phone)
                .having(func.count(Lead.id) > 1)
                .subquery()
            )
            .scalar()
            or 0
        )

        empty_names = (
            self.db.query(func.count(Lead.id))
            .filter((Lead.name.is_(None)) | (Lead.name == ""))
            .scalar()
            or 0
        )

        return {
            "total_records": total_leads + total_properties + total_users,
            "leads": total_leads,
            "properties": total_properties,
            "users": total_users,
            "quick_checks": {
                "duplicate_phones": duplicate_phones,
                "empty_names": empty_names,
            },
        }
